package com.ruoyi.framework;

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * MQTT监听服务类
 * 用于监听不同的MQTT主题，并根据主题调用不同的方法处理消息
 */
@Service
public class MqttListenerService {

    private static final Logger logger = LoggerFactory.getLogger(MqttListenerService.class);

    // MQTT客户端实例
    private MqttClient client;
    
    // 存储主题与消息处理器的映射关系
    private Map<String, MessageHandler> topicHandlers = new HashMap<>();
    
    // 线程池，用于异步处理消息
    private ExecutorService executorService;
    
    // 数据库服务
    @Autowired
    private DatabaseService databaseService;
    
    // MQTT配置
    @Autowired
    private MqttConfig mqttConfig;
    
    /**
     * 初始化MQTT客户端并连接到 broker
     */
    @PostConstruct
    public void init() {
        try {
            // 创建线程池
            executorService = Executors.newFixedThreadPool(10);
            
            // 创建MQTT客户端
            client = new MqttClient(mqttConfig.getBroker(), mqttConfig.getClientId(), new MemoryPersistence());
            
            // 设置连接选项
            MqttConnectOptions options = new MqttConnectOptions();
            options.setUserName(mqttConfig.getUsername());
            options.setPassword(mqttConfig.getPassword().toCharArray());
            options.setAutomaticReconnect(mqttConfig.isAutomaticReconnect());
            options.setCleanSession(mqttConfig.isCleanSession());
            options.setConnectionTimeout(mqttConfig.getConnectionTimeout());
            options.setKeepAliveInterval(mqttConfig.getKeepAliveInterval());
            
            // 设置回调
            client.setCallback(new MqttCallback() {
                @Override
                public void connectionLost(Throwable cause) {
                    logger.error("MQTT连接丢失: {}", cause.getMessage());
                }
                
                @Override
                public void messageArrived(String topic, MqttMessage message) throws Exception {
                    logger.info("接收到消息 - 主题: {}, 内容: {}", topic, new String(message.getPayload()));
                    
                    // 异步处理消息
                    executorService.submit(() -> {
                        try {
                            // 根据主题调用对应的处理器
                            MessageHandler handler = topicHandlers.get(topic);
                            if (handler != null) {
                                handler.handle(topic, new String(message.getPayload()));
                            } else {
                                logger.warn("未找到主题 \"{}\" 对应的处理器", topic);
                            }
                        } catch (Exception e) {
                            logger.error("处理消息失败: {}", e.getMessage(), e);
                        }
                    });
                }
                
                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {
                    // 消息发送完成回调，这里不做处理
                }
            });
            
            // 连接到MQTT服务器
            client.connect(options);
            logger.info("MQTT客户端连接成功");
            
            // 初始化消息处理器
            initMessageHandlers();
            
            // 订阅主题
            subscribeTopics();
            
        } catch (MqttException e) {
            logger.error("MQTT客户端初始化失败: {}", e.getMessage(), e);
        }
    }
    
    /**
     * 初始化消息处理器
     */
    private void initMessageHandlers() {
        // 注册不同主题的处理器
        topicHandlers.put("device/data", this::handleDeviceData);
        topicHandlers.put("user/command", this::handleUserCommand);
        topicHandlers.put("system/notification", this::handleSystemNotification);
        
        // 可以根据需要注册更多的主题处理器
    }
    
    /**
     * 订阅所有已注册处理器的主题
     */
    private void subscribeTopics() { 
        try {
            // 使用配置的QoS级别
            int qos = mqttConfig.getDefaultQos();
            
            // 订阅所有已注册的主题
            for (String topic : topicHandlers.keySet()) {
                client.subscribe(topic, qos);
                logger.info("已订阅主题: {}", topic);
            }
        } catch (MqttException e) {
            logger.error("订阅主题失败: {}", e.getMessage(), e);
        }
    }
    
    /**
     * 发布消息到MQTT服务器
     * @param topic 主题
     * @param message 消息内容
     * @param qos QoS级别
     */
    public void publish(String topic, String message, int qos) {
        try {
            if (client != null && client.isConnected()) {
                MqttMessage mqttMessage = new MqttMessage(message.getBytes());
                mqttMessage.setQos(qos);
                client.publish(topic, mqttMessage);
                logger.info("发布消息成功 - 主题: {}, 内容: {}", topic, message);
            } else {
                logger.error("MQTT客户端未连接，无法发布消息");
            }
        } catch (MqttException e) {
            logger.error("发布消息失败: {}", e.getMessage(), e);
        }
    }
    
    /**
     * 发布消息到MQTT服务器（使用默认QoS级别）
     * @param topic 主题
     * @param message 消息内容
     */
    public void publish(String topic, String message) {
        publish(topic, message, mqttConfig.getDefaultQos());
    }
    
    /**
     * 处理设备数据消息
     */
    private void handleDeviceData(String topic, String message) {
        logger.info("处理设备数据: {}", message);
        try {
            // 调用数据库服务保存设备数据
            databaseService.saveDeviceData(message);
        } catch (Exception e) {
            logger.error("保存设备数据失败: {}", e.getMessage(), e);
        }
    }
    
    /**
     * 处理用户命令消息
     */
    private void handleUserCommand(String topic, String message) {
        logger.info("处理用户命令: {}", message);
        try {
            // 调用数据库服务处理用户命令
            databaseService.processUserCommand(message);
        } catch (Exception e) {
            logger.error("处理用户命令失败: {}", e.getMessage(), e);
        }
    }
    
    /**
     * 处理系统通知消息
     */
    private void handleSystemNotification(String topic, String message) {
        logger.info("处理系统通知: {}", message);
        try {
            // 调用数据库服务保存系统通知
            databaseService.saveSystemNotification(message);
        } catch (Exception e) {
            logger.error("保存系统通知失败: {}", e.getMessage(), e);
        }
    }
    
    /**
     * 断开MQTT连接并清理资源
     */
    @PreDestroy
    public void destroy() {
        try {
            if (client != null && client.isConnected()) {
                client.disconnect();
                logger.info("MQTT客户端已断开连接");
            }
            
            if (executorService != null && !executorService.isShutdown()) {
                executorService.shutdown();
                logger.info("线程池已关闭");
            }
        } catch (MqttException e) {
            logger.error("MQTT客户端关闭失败: {}", e.getMessage(), e);
        }
    }
    
    /**
     * 消息处理器接口
     */
    @FunctionalInterface
    private interface MessageHandler {
        void handle(String topic, String message);
    }
}